package com.mangix.core;

import com.mangix.conf.MangixConfig;
import com.mangix.http.MangixHttpClient;
import com.mangix.http.MangixMonitorServlet;
import com.mangix.utils.MangixUtils;
import org.eclipse.jetty.client.HttpClient;
import org.eclipse.jetty.client.api.Request;
import org.eclipse.jetty.client.api.Response;
import org.eclipse.jetty.client.api.Result;
import org.eclipse.jetty.client.util.StringContentProvider;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.ServerConnector;
import org.eclipse.jetty.servlet.ServletHandler;
import org.eclipse.jetty.servlet.ServletHolder;
import org.json.simple.JSONValue;

import java.io.IOException;
import java.io.OutputStreamWriter;
import java.net.HttpURLConnection;
import java.net.MalformedURLException;
import java.net.URL;
import java.net.URLConnection;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;


/**
 * Created by Ahan on 2014/5/28.
 */
public class MangixMonitor {

    private int M;
    private long RING_SIZE;
    private MangixConfig config;
    private MangixMonitor superior;
    private String hostname;
    private int port;
    private String state = "Normal";

    // A monitor instance can be either local or remote.
    private boolean isLocal;
    private Map<Long, Map<String, Object>> nodes;

    public MangixMonitor(MangixConfig config) {
        this.config = config;
        this.hostname = config.getHostName();
        this.port = config.getMonitorPort();
        this.nodes = new ConcurrentHashMap<>();
        this.M = config.getM();
        RING_SIZE = (long) Math.pow(2, M);
        this.isLocal = true;
    }

    public MangixMonitor(String hostname, int port) {
        this.hostname = hostname;
        this.port = port;
        this.isLocal = false;
    }

    public String getHostname() {
        return this.hostname;
    }

    public int getPort() {
        return this.port;
    }

    public void update(String reportContent) {
        Map nodeInfo = (Map) JSONValue.parse(reportContent);
        String hostname = nodeInfo.get("hostname").toString();
        int port = ((Number)nodeInfo.get("port")).intValue();
        long id = MangixUtils.hash(hostname + ":" + port, M);
        nodeInfo.put("is_active", true);
        this.nodes.put(id, nodeInfo);
    }

    /**
     * Check whether the processor, successor, and finger table of each node are correct.
     * To do this, we have to ensure that all the nodes should report their info to the same monitor.
     */
    private void check() {
        List<Map.Entry<Long, Map<String, Object>>> livingNodes = null;
        // Count for number of nodes;
        livingNodes = new ArrayList(nodes.entrySet());
        Collections.sort(livingNodes, new Comparator<Map.Entry<Long, Map<String, Object>>>() {
            @Override
            public int compare(Map.Entry<Long, Map<String, Object>> o1, Map.Entry<Long, Map<String, Object>> o2) {
                return o1.getKey().compareTo(o2.getKey());
            }
        });
        int size = livingNodes.size();
        System.out.print("There are " + size + " nodes now:");
        boolean isOK = true;
        for (int i = 0; i < size; i++) {
            Map<String, Object> currentNode, processorInList, successorInList;
            currentNode = livingNodes.get(i).getValue();
            if (i - 1 < 0) {
                processorInList = livingNodes.get(size - 1).getValue();
            } else {
                processorInList = livingNodes.get(i - 1).getValue();
            }
            successorInList = livingNodes.get((i + 1) % size).getValue();

            String currentNodeHostname = (String)currentNode.get("hostname");
            int currentNodePort = ((Number)currentNode.get("port")).intValue();
            String processorHostname = null;
            int processorPort = 0;
            Map processor = (Map)currentNode.get("processor");
            if (processor != null) {
                processorHostname = (String)processor.get("hostname");
                if (processor.get("port") != null) {
                    processorPort = ((Number)processor.get("port")).intValue();
                }
            }
            String successorHostname = null;
            int successorPort = 0;
            Map successor = (Map)currentNode.get("successor");
            if (successor != null) {
                successorHostname = (String)successor.get("hostname");
                if (processor.get("port") != null) {
                    successorPort = ((Number)successor.get("port")).intValue();
                }
            }
            if (processorHostname == null || processorPort == 0 || successorHostname == null || successorPort == 0) {
                System.out.println("The processor and successor of node " + currentNodeHostname + ":" + currentNodePort + " is incomplete.");
                isOK = false;
                break;
            }

            if (!processorHostname.equals(processorInList.get("hostname")) || processorPort != ((Number)processorInList.get("port")).intValue()) {
                isOK = false;
                System.out.println("Current node " + currentNodeHostname + ":" + currentNodePort + "'s processor is:" + processorHostname + ":" + processorPort + ", but in fact its processor should be :" + (String)processorInList.get("hostname") + ":" + processorInList.get("port").toString());
                break;
            }

            if (!successorHostname.equals(successorInList.get("hostname")) || successorPort != ((Number)successorInList.get("port")).intValue()) {
                isOK = false;
                System.out.println("Current node:" + currentNodeHostname + ":" + currentNodePort + "'s successor is:" + successorHostname + ":" + successorPort + ", but in fact its successor should be:" + (String)successorInList.get("hostname") + ":" + successorInList.get("port").toString());
                break;
            }

            // Check the finger table.
            List fingerTable = (List)currentNode.get("finger_table");
            long currentId = livingNodes.get(i).getKey();
            for (int k = 0; k < M; k++) {
                Map finger = (Map) fingerTable.get(k);
                long start = ((Number)finger.get("start")).longValue();
                if (start != (long) (currentId + Math.pow(2, k)) % RING_SIZE) {
                    isOK = false;
                    System.out.println("The " + k + " th start of node " + currentNodeHostname + ":" + currentNodePort+ " is not correct.");
                    break;
                }
                String coordinatorHostname = (String) finger.get("hostname");
                int coordinatorPort = ((Number) finger.get("port")).intValue();
                if (coordinatorHostname == null || coordinatorPort == 0) {
                    isOK = false;
                    System.out.println("Node " + currentNodeHostname + ":" + currentNodePort +"'s finger is null.");
                    break;
                }
                for (int j = 0; j < size; j++) {
                    Map otherNode = livingNodes.get(j).getValue();
                    String otherNodeName = otherNode.get("hostname") + ":" + otherNode.get("port").toString();
                    long otherNodeId = MangixUtils.hash(otherNodeName, M);
                    long coordinatorId = MangixUtils.hash(coordinatorHostname + ":" + coordinatorPort, M);
                    if (otherNodeId != coordinatorId && start != coordinatorId) {
                        if (MangixUtils.isIDInInterval(otherNodeId, start, coordinatorId, true, false, M)) {
                            isOK = false;
                            System.out.println("The " + k + "th finger of node " + currentNodeHostname + ":" + currentNodePort + "(" + currentId + ") is not correct.");
                            System.out.println("The coordinator of start " + start + " is now " + coordinatorHostname + ":" + coordinatorPort + "(" + coordinatorId + ")" + " ,but it should be " + otherNodeName + "(" + otherNodeId + ")");
                            break;
                        }
                    }
                }
            }

        }
        if (isOK) {
            state = "Normal";
        } else {
            state = "Rebuilding";
        }
        System.out.println("Is the Mangix system in right state? " + isOK);
    }

    public void setMonitor(MangixMonitor monitor) {
        this.superior = monitor;
    }

    public void addObservation(String hostname, int port) {
        Map obj = new LinkedHashMap();
        obj.put("hostname", this.hostname);
        obj.put("port", new Integer(this.port));
        String jsonText = JSONValue.toJSONString(obj);
        try {
            MangixHttpClient.put("http://" + hostname + ":" + port + "/monitor", jsonText);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * Start the mangix.Mangix monitor.
     */
    public void start() {
        Server server = new Server();
        ServerConnector http = new ServerConnector(server);
        http.setHost(hostname);
        http.setPort(port);
        http.setAcceptQueueSize(100);
        server.addConnector(http);
        ServletHandler handler = new ServletHandler();
        server.setHandler(handler);
        handler.addServletWithMapping(new ServletHolder(new MangixMonitorServlet(this)), "/");

        try {
            server.start();
            new Thread(new Runnable() {
                @Override
                public void run() {
                while(true) {
                    try {
                        for (Map nodeInfo : nodes.values()) {
                            String hostname = (String) nodeInfo.get("hostname");
                            int port = ((Number)nodeInfo.get("port")).intValue();
                            MangixNode node = MangixNode.createNode(hostname, port);
                            if (!node.isActive()) {
                                long nodeId = MangixUtils.hash(hostname + ":" + port, M);
                                nodes.remove(nodeId);
                            }
                        }
                        Thread.sleep(1000 * 30);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                }
            }).start();

            new Thread(new Runnable() {
                @Override
                public void run() {
                    while(true) {
                        try {
                            Thread.sleep(1000 * 10);
                        } catch (InterruptedException e1) {
                            e1.printStackTrace();
                        }
                        try {
                            check();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
            }).start();
            System.out.println("Mangix Monitor is running on " + this.toString() + ".");
        } catch (Exception e) {
            e.printStackTrace();
            System.exit(-1);
        }

    }

    public void report(Map<String, Object> nodeInfo) {

        String jsonText = JSONValue.toJSONString(nodeInfo);
        try {
            MangixHttpClient.put("http://" + this.hostname + ":" + this.port + "/report", jsonText);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public List<Map<String, Object>> getNodes() {
        List<Map<String, Object>> nodeList = new LinkedList<>();
        for (Long key : nodes.keySet()) {
            nodeList.add(nodes.get(key));
        }
        return nodeList;
    }

    public String getState() {
        return state;
    }

    @Override
    public String toString() {
        return hostname + ":" + port;
    }

    public static void main(String[] args) {
        MangixMonitor monitor = new MangixMonitor("192.168.56.111", 9093);
        Map nodeInfo = new LinkedHashMap();
        nodeInfo.put("hostname", "192.168.56.111");
        nodeInfo.put("port", new Integer(8080));
        nodeInfo.put("used_space", new Integer(13));
        monitor.report(nodeInfo);
    }

}
